/******************************************************************************
 *  author: enquanyan
 *  version: 1.0
 *  description:
 *  			threadpoll header file;
 *  			provide thread pool;
 *  date: 2014-07-21
 *  other: 无
 ******************************************************************************/

#ifndef THREADPOOL_H
#define THREADPOOL_H

#include <list>
#include <cstdio>
#include <exception>
#include <pthread.h>
#include "locker.h"
#include "thread_arg.h"
#include "def.h"

template<typename T>
class threadpool {
public:
	threadpool(int thread_number = 8, int max_requests = 10000);
	~threadpool();
	bool append(T* request);
	bool append(T* request, int thread_no);
	
	int judge_relax_thread(void);
	
	static thread_arg *thread_arg_array;

private:
	static void* worker(void* arg);
	void run(int thread_no);

private:
	int m_thread_number;
	int m_max_requests;
	pthread_t* m_threads;
	std::list<T*> m_workqueue;
	locker m_queuelocker;
	sem m_queuestat;
	static bool m_stop;
};

template<typename T>
thread_arg* threadpool<T>::thread_arg_array = NULL;

template<typename T>
bool threadpool<T>::m_stop = false;

template<typename T>
threadpool<T>::threadpool(int thread_number, int max_requests) :
		m_thread_number(thread_number), m_max_requests(max_requests), m_threads( NULL) {
	if ((thread_number <= 0) || (max_requests <= 0)) {
		throw std::exception();
	}
	
	thread_arg_array = new thread_arg[m_thread_number];
	
	m_threads = new pthread_t[m_thread_number];
	
	if (!m_threads) {
		throw std::exception();
	}
	 
	for (int i = 0; i < thread_number; ++i) {
		
		thread_arg_array[i].thread_no = i;
		thread_arg_array[i].thread_pool_instance = this;
		thread_arg_array[i].active_http_conn = 0;
		thread_arg_array[i].interval_received_bytes_first = 0;
		thread_arg_array[i].interval_received_bytes_second = 0;
		thread_arg_array[i].interval_handled_http_conn_first = 0;
		thread_arg_array[i].interval_handled_http_conn_second = 0;
		thread_arg_array[i].is_first = true;
		
		thread_arg_array[i].is_handling = false;
		thread_arg_array[i].is_stop = false;
		
		printf("create the %dth thread\n", i);
		
		if (pthread_create(m_threads + i, NULL, worker, &(thread_arg_array[i].thread_no)) != 0)
		{
			delete[] m_threads;
			throw std::exception();
		}
		
		if (pthread_detach(m_threads[i]))
		{
			delete[] m_threads;
			throw std::exception();
		}
	}
}

template<typename T>
threadpool<T>::~threadpool() {
	delete[] thread_arg_array;
	delete[] m_threads;
	m_stop = true;
}

template<typename T>
bool threadpool<T>::append(T* request) {
	m_queuelocker.lock();
	if (m_workqueue.size() > m_max_requests) {
		m_queuelocker.unlock();
		return false;
	}
	m_workqueue.push_back(request);
	m_queuelocker.unlock();
	m_queuestat.post();
	return true;
}

template<typename T>
bool threadpool<T>::append(T* request, int thread_no) 
{
	if(thread_arg_array[thread_no].http_conn_queue.push((void*)request))
	{
		thread_arg_array[thread_no].queue_stat.post();
		
		return true;
	}

	return false;
}

template<typename T>
void* threadpool<T>::worker(void* arg) {
	int thread_no = *(int*)arg;
	threadpool<T>* pool = (threadpool<T>* )(thread_arg_array[thread_no].thread_pool_instance);
	pool->run(thread_no);
	return pool;
}

template<typename T>
void threadpool<T>::run(int thread_no)
{
	thread_arg *thread_arg_pointer = thread_arg_array + thread_no;
	
	while (!m_stop && !thread_arg_pointer->is_stop) 
	{
		thread_arg_pointer->queue_stat.wait();
		
		void *temp = NULL;
		if(thread_arg_pointer->http_conn_queue.pop(&temp))
		{
			T* request = (T*)temp;
			thread_arg_pointer->is_handling = true;
            
            //pr_debug("into request->process\n");
            
			request->process();
            
			thread_arg_pointer->is_handling = false;
		}
		
		continue;
	}
}

template<typename T>
int threadpool<T>::judge_relax_thread()
{
	unsigned int thread_no = 0;
	for(int i= 1 ; i <  m_thread_number; i++)
	{
		if(thread_arg_array[i].active_http_conn < thread_arg_array[thread_no]. active_http_conn)
		{
			thread_no = i;
		}
	}
	return thread_no;
}

#endif
